package com.tj.bigdata.syncblock.service.impl;

import com.alibaba.fastjson.JSON;
import com.tj.bigdata.syncblock.config.ElasticSearchConfig;
import com.tj.bigdata.syncblock.constant.EsConstant;
import com.tj.bigdata.syncblock.constant.IndexConstant;
import com.tj.bigdata.syncblock.entity.BlockInfo;
import com.tj.bigdata.syncblock.entity.Transactions;
import com.tj.bigdata.syncblock.service.EsUpService;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.web3j.protocol.core.methods.response.EthBlock;

import java.io.IOException;
import java.math.BigInteger;
import java.util.List;

@Slf4j
@Service
public class EsUpServiceImpl implements EsUpService {

    @Autowired
    private RestHighLevelClient restHighLevelClient;

    @Value("${spring.syncblock.index}")
    private String index;

    @Override
    public Boolean blockInfoUp(BlockInfo block) throws IOException {
        log.info("==========区块信息开始上传ES {}====》", block.getNumber());
        BulkRequest bulkRequest = new BulkRequest();
        IndexRequest indexRequest = new IndexRequest(EsConstant.BLOCK_INFO + IndexConstant.INTERVAL_SYMBOL + index + IndexConstant.INTERVAL_SYMBOL + IndexConstant.STEP_SIZE_MARK_50W);
        indexRequest.id(block.getNumber().toString());
        String s = JSON.toJSONString(block);
        indexRequest.source(s, XContentType.JSON);
        bulkRequest.add(indexRequest);
        BulkResponse bulk = restHighLevelClient.bulk(bulkRequest, ElasticSearchConfig.COMMON_OPTIONS);
        return !bulk.hasFailures();
    }

    @Override
    public Boolean blockInfoUp(EthBlock.Block block) throws IOException {
        log.info("==========区块信息开始上传ES {}====》", block.getNumber());
        BulkRequest bulkRequest = new BulkRequest();
        String blockIndex = EsConstant.BLOCK_INFO + IndexConstant.INTERVAL_SYMBOL + index + IndexConstant.INTERVAL_SYMBOL + IndexConstant.STEP_SIZE_MARK_50W;
        log.info("上传区块的索引为 {}", blockIndex);
        IndexRequest indexRequest = new IndexRequest(blockIndex);
        indexRequest.id(block.getNumber().toString());
        block.setTransactions(null);
        String s = JSON.toJSONString(block);
        indexRequest.source(s, XContentType.JSON);
        bulkRequest.add(indexRequest);
        BulkResponse bulk = restHighLevelClient.bulk(bulkRequest, ElasticSearchConfig.COMMON_OPTIONS);
        return !bulk.hasFailures();
    }

    @Override
    public Boolean transactionInfoUp(List<Transactions> transactions) throws IOException {
        BulkRequest bulkRequest = new BulkRequest();
        String transactionIndex = EsConstant.TRANSACTION_INFO + IndexConstant.INTERVAL_SYMBOL + index + IndexConstant.INTERVAL_SYMBOL + IndexConstant.STEP_SIZE_MARK_50W;
        log.info("上传交易的索引为 {}", transactionIndex);
        for (Transactions transaction : transactions) {
            IndexRequest indexRequest = new IndexRequest(transactionIndex);
            indexRequest.id(transaction.getHash());
            String s = JSON.toJSONString(transaction);
            indexRequest.source(s, XContentType.JSON);
            bulkRequest.add(indexRequest);
        }
        BulkResponse bulk = restHighLevelClient.bulk(bulkRequest, ElasticSearchConfig.COMMON_OPTIONS);
        return !bulk.hasFailures();
    }

    @Override
    public Boolean transactionInfoStreamSourceUp(List<Transactions> transactions) throws IOException {
        BulkRequest bulkRequest = new BulkRequest();
        String transactionIndex = EsConstant.TRANSACTION_INFO;
        for (Transactions transaction : transactions) {
            IndexRequest indexRequest = new IndexRequest(transactionIndex);
            indexRequest.id(transaction.getHash());
            String s = JSON.toJSONString(transaction);
            indexRequest.source(s, XContentType.JSON);
            bulkRequest.add(indexRequest);
        }
        BulkResponse bulk = restHighLevelClient.bulk(bulkRequest, ElasticSearchConfig.COMMON_OPTIONS);
        return !bulk.hasFailures();
    }

    @Override
    public Boolean transactionInfoUp(Transactions transaction) throws IOException {
        BulkRequest bulkRequest = new BulkRequest();
        IndexRequest indexRequest = new IndexRequest(EsConstant.TRANSACTION_INFO);
        indexRequest.id(transaction.getHash());
        String s = JSON.toJSONString(transaction);
        indexRequest.source(s, XContentType.JSON);
        bulkRequest.add(indexRequest);
        BulkResponse bulk = restHighLevelClient.bulk(bulkRequest, ElasticSearchConfig.COMMON_OPTIONS);
        return !bulk.hasFailures();
    }

    @Override
    public BigInteger maxBlockBumber() {
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.sort("number", SortOrder.DESC);
        searchSourceBuilder.size(1);
        searchSourceBuilder.query(QueryBuilders.matchAllQuery());
        searchSourceBuilder.trackTotalHits(true);
        SearchRequest searchRequest = new SearchRequest(new String[]{EsConstant.BLOCK_INFO}, searchSourceBuilder);
        log.info(">>>>>>>>>>>构建的查询最大同步区块number DSL语句----{}<<<<<<<<<<<<<<<<", searchSourceBuilder.toString());
        try {
            SearchResponse response = restHighLevelClient.search(searchRequest, ElasticSearchConfig.COMMON_OPTIONS);
            SearchHits hits = response.getHits();
            if (hits != null && hits.getHits().length > 0) {
                return BigInteger.valueOf(hits.getTotalHits().value);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        log.info("==========当前同步最大区块信息获取失败 ======");
        return null;
    }

}
